package com.qys.livesMall.kafka.send;

import com.qys.livesMall.kafka.entity.CallBackExpressVO;
import com.qys.livesMall.kafka.entity.log.SystemLog;
import com.qys.livesMall.kafka.entity.shop.ShopBrowseRecord;
import com.qys.livesMall.kafka.enums.QysTopic;
import com.qys.livesMall.kafka.handle.KafkaSendResultHandler;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author 徐一杰
 * @date 2022/10/31 14:05
 * kafka发送消息
 */
@Component
public class KafkaClient {

    private final KafkaTemplate<Object, Object> kafkaTemplate;

    @Autowired
    private QysTopic qysTopic;

    public KafkaClient(KafkaTemplate<Object, Object> kafkaTemplate, KafkaSendResultHandler kafkaSendResultHandler) {
        this.kafkaTemplate = kafkaTemplate;
        //回调方法、异常处理
        this.kafkaTemplate.setProducerListener(kafkaSendResultHandler);
    }

    public void shopBrowseRecord(ShopBrowseRecord record) {
        String message = "发送到Kafka的消息";
        String topic = qysTopic.shop_browse_records;
        kafkaTemplate.send(qysTopic.shop_browse_records, record);


    }

    /**
     * 后台消息日志
     *
     * @param record {@link SystemLog}
     */
    public void systemLog(SystemLog record) {
        // kafkaTemplate.send(qysTopic.systemLogRecords, record);
    }


    public void send() {
        //这个User的代码我没放出来，自己随便写一个实体类，实体类一定要 implements Serializable

        kafkaTemplate.send("topic2", "发给topic2");
    }

    /**
     * Kafka提供了多种构建消息的方式
     *
     * @throws ExecutionException
     * @throws InterruptedException
     * @throws TimeoutException
     */
    public void SendDemo() throws ExecutionException, InterruptedException, TimeoutException {
        //后面的get代表同步发送，括号内时间可选，代表超过这个时间会抛出超时异常，但是仍会发送成功
        kafkaTemplate.send("topic1", "发给topic1").get(1, TimeUnit.MILLISECONDS);

        //使用ProducerRecord发送消息
        ProducerRecord<Object, Object> producerRecord = new ProducerRecord<>("topic.quick.demo", "use ProducerRecord to send message");
        kafkaTemplate.send(producerRecord);

        //使用Message发送消息
        Map<String, Object> map = new HashMap<>();
        map.put(KafkaHeaders.TOPIC, "topic.quick.demo");
        GenericMessage<Object> message = new GenericMessage<>("use Message to send message", new MessageHeaders(map));
        kafkaTemplate.send(message);
    }

    /**
     * 快递消息
     * @param express ExpressInquiryVO
     */
    public void expressMessage(List<CallBackExpressVO> express) {
        kafkaTemplate.send(qysTopic.expressMessage,express);
    }
}
